package com.lboyang.remoting.transport.netty.server;

import com.lboyang.common.enums.CompressTypeEnum;
import com.lboyang.common.enums.RpcResponseCodeEnum;
import com.lboyang.common.enums.SerializationTypeEnum;
import com.lboyang.common.factory.SingletonFactory;
import com.lboyang.common.constants.NrpcConstants;
import com.lboyang.common.dto.NrpcMessage;
import com.lboyang.common.dto.NrpcRequest;
import com.lboyang.common.dto.NrpcResponse;
import com.lboyang.remoting.transport.netty.handler.RpcRequestHandler;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.ReferenceCountUtil;
import lombok.extern.slf4j.Slf4j;

/**
 * @Description
 *      服务器端处理器，处理方法并返回给客户端
 * @Date 2021/9/12 16:34
 * @author: A.iguodala
 */
@Slf4j
public class NettyRpcServerHandler extends ChannelInboundHandlerAdapter {


    /**
     * 请求的本地方法处理器
     */
    private final RpcRequestHandler rpcRequestHandler;

    public NettyRpcServerHandler() {
        this.rpcRequestHandler = SingletonFactory.getInstance(RpcRequestHandler.class);
    }

    /**
     * 处理读事件
     * @param ctx
     * @param msg
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        try {
            if (msg instanceof NrpcMessage) {
                log.info("server receive msg: [{}] ", msg);
                byte messageType = ((NrpcMessage) msg).getMessageType();
                NrpcMessage rpcMessage = new NrpcMessage();
                rpcMessage.setCodec(SerializationTypeEnum.PROTOSTUFF.getCode());
                rpcMessage.setCompress(CompressTypeEnum.GZIP.getCode());
                if (messageType == NrpcConstants.HEARTBEAT_REQUEST_TYPE) {
                    rpcMessage.setMessageType(NrpcConstants.HEARTBEAT_RESPONSE_TYPE);
                    rpcMessage.setData(NrpcConstants.PONG);
                } else {
                    NrpcRequest rpcRequest = (NrpcRequest) ((NrpcMessage) msg).getData();
                    // 真正调用方法执行处理
                    Object result = rpcRequestHandler.handle(rpcRequest);
                    log.info(String.format("server get result: %s", result.toString()));
                    rpcMessage.setMessageType(NrpcConstants.RESPONSE_TYPE);
                    if (ctx.channel().isActive() && ctx.channel().isWritable()) {
                        NrpcResponse<Object> rpcResponse = NrpcResponse.success(result, rpcRequest.getRequestId());
                        rpcMessage.setData(rpcResponse);
                    } else {
                        NrpcResponse<Object> rpcResponse = NrpcResponse.fail(RpcResponseCodeEnum.FAIL);
                        rpcMessage.setData(rpcResponse);
                        log.error("not writable now, message dropped");
                    }
                }
                ctx.writeAndFlush(rpcMessage).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
            }
        } finally {
            //确保空间释放
            ReferenceCountUtil.release(msg);
        }
    }

    /**
     * 没有写事件就关闭
     * @param ctx
     * @param evt
     * @throws Exception
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleState state = ((IdleStateEvent) evt).state();
            if (state == IdleState.READER_IDLE) {
                log.info("idle check happen, so close the connection");
                ctx.close();
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }

    /**
     * 异常处理
     * @param ctx
     * @param cause
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        log.error("server catch exception");
        cause.printStackTrace();
        ctx.close();
    }
}
